package com.sise.cn.task.config;

import com.sise.cn.task.enumerate.QueueEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;

/**
 * @author 龚秀峰
 * @version 1.0
 * @date 2020/6/7 21:24
 * 消息队列的配置
 */
@Configuration
@Slf4j
public class RabbitMqConfig {

    @Bean
    public RabbitListenerContainerFactory<SimpleMessageListenerContainer> rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter converter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(converter);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //开启退回模式，确保能消息能投送到队列中
        rabbitTemplate.setUsePublisherConnection(true);
        //强制开启委托模式
        rabbitTemplate.setMandatory(true);
        //设置确认回调
        rabbitTemplate.setConfirmCallback(this::confirm);
        //设置异常回调
        rabbitTemplate.setReturnCallback(this::returnedMessage);
        return rabbitTemplate;
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 将要被执行的任务队列所绑定的交换机
     */
    @Bean
    public DirectExchange taskDirectExchange() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_TASK.getExchange())
                .durable(true)
                .build();
    }

    /**
     * 延时任务队列所绑定的交换机
     */
    @Bean
    public DirectExchange taskTtlDirectExchange() {
        return (DirectExchange) ExchangeBuilder
                .directExchange(QueueEnum.QUEUE_TASK_TTL.getExchange())
                .durable(true)
                .build();
    }

    /**
     * 将要被执行的任务队列
     */
    @Bean
    public Queue taskQueue() {
        return new Queue(QueueEnum.QUEUE_TASK.getName());
    }

    /**
     * 延时执行任务的队列
     * 当延时结束，任务会被发送到将要执行任务的队列中执行
     * 例如：超时未付款，则发送到取消订单的队列
     */
    @Bean
    public Queue taskTtlQueue() {
        return QueueBuilder
                .durable(QueueEnum.QUEUE_TASK_TTL.getName())
                .withArgument("x-dead-letter-exchange", QueueEnum.QUEUE_TASK.getExchange())
                .withArgument("x-dead-letter-routing-key", QueueEnum.QUEUE_TASK.getRouteKey())
                .build();
    }

    /**
     * 将要执行的任务队列绑定到交换机
     */
    @Bean
    public Binding taskBinding(DirectExchange taskDirectExchange, Queue taskQueue) {
        return BindingBuilder
                .bind(taskQueue)
                .to(taskDirectExchange)
                .with(QueueEnum.QUEUE_TASK.getRouteKey());
    }

    /**
     * 延时任务队列绑定到交换机
     */
    @Bean
    public Binding taskTtlBinding(DirectExchange taskTtlDirectExchange, Queue taskTtlQueue) {
        return BindingBuilder
                .bind(taskTtlQueue)
                .to(taskTtlDirectExchange)
                .with(QueueEnum.QUEUE_TASK_TTL.getRouteKey());
    }

    /**
     * 消息确认的回调函数
     *
     * @param correlationData 消息描述对象
     * @param ack             broker是否成功应答
     * @param cause           未应答的原因
     */
    private void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("当前已经确认发送的消息ID为：" + correlationData.getId());
        } else {
            log.info("当前已经发送暂未确认的消息ID为：" + correlationData.getId() + "原因是：" + cause);
        }
    }

    /**
     * 消息发送失败的回调函数
     *
     * @param message    消息对象
     * @param replyCode  失败码
     * @param replyText  失败信息
     * @param exchange   交换机
     * @param routingKey 路由key
     */
    private void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("当前发送失败的消息为：" + Arrays.toString(message.getBody()));
    }

}
